package com.geezdata.cps.cqrs.event.publish.impl;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.geezdata.cps.cqrs.event.Event;
import com.geezdata.cps.cqrs.event.annotation.EventPublisher;
import com.geezdata.cps.cqrs.event.enums.EventSource;
import com.geezdata.cps.cqrs.event.listener.Listener;
import com.geezdata.cps.cqrs.event.listener.container.ListenerContainer;
import com.geezdata.cps.cqrs.event.publish.AbstractPublisher;


@EventPublisher(source = EventSource.ASYNC)
public class DefaultAsyncPublisher extends AbstractPublisher {
	private static Logger logger = LoggerFactory.getLogger(DefaultAsyncPublisher.class);
	
    private static final int MAX_CONSUMER = 5;
    private ExecutorService consumer;
    public DefaultAsyncPublisher(ListenerContainer listenerContainer) {
        super(listenerContainer);
        consumer = Executors.newFixedThreadPool(MAX_CONSUMER);
    }
    
    public DefaultAsyncPublisher(ListenerContainer listenerContainer, Integer maxConusmer) {
        super(listenerContainer);
        consumer = Executors.newFixedThreadPool(maxConusmer);
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
	@Override
    public <T> void publish(Event<T> event) {
    	consumer.execute(() -> {
            List<Listener> listeners = listenerContainer.getListeners(event.getClass());
            if (listeners != null) {
                for (Listener listener : listeners) {
                    try {
                        listener.onEvent(event);
                    } catch (Exception e) {
                        logger.error("AsyncPublisher catch exception. {}", e);
                    }
                }
            }
        });
    }

    public void stop() {
        try {
            consumer.shutdown();
            if (!consumer.awaitTermination(20, TimeUnit.SECONDS)) {
                logger.error("Await termination timeout.");
            }
        } catch (InterruptedException e) {
            logger.error("Catch interrupted exception. ", e);
        }
    }
}
